package com.tengju.data.infrastructure.kafka.config;

import com.tengju.data.infrastructure.kafka.consumer.HkUserBinlogConsumer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @author chenhang
 * @Date 2021/9/29
 */
@Configuration
public class HkUserBinlogKafkaConfig {

    @Value("${hkbinlog.kafka.consumer.bootstrapServers}")
    private String bootstrapServers;

    @Value("${hkbinlog.kafka.consumer.user_brand_group_name}")
    private String userBrandGroupName;

    @Value("${hkbinlog.kafka.consumer.intervalMs}")
    private int intervalMs;

    @Value("${hkbinlog.kafka.consumer.timeoutMs}")
    private int timeoutMs;

    @Value("${hkbinlog.kafka.consumer.autoOffset}")
    private String autoOffset;

    @Value("${hkbinlog.kafka.consumer.keyDeserializer}")
    private String keyDeserializer;

    @Value("${hkbinlog.kafka.consumer.valueDeserializer}")
    private String valueDeserializer;

    @Value("${hkbinlog.kafka.consumer.autoCommit}")
    private boolean autoCommit;

    @Value("${hkbinlog.kafka.consumer.pollRecords}")
    private int pollRecords;

    @Value("${hkbinlog.kafka.consumer.user_brand_provider_topic}")
    private String userBrandProviderTopic;

    @Value("${hkbinlog.consumer.swich}")
    private boolean switchFlag;


    @Bean("kafkaUserInfoConsumerProperties")
    public Properties kafkaUserInfoConsumerProperties() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("group.id", userBrandGroupName);
        properties.put("auto.commit.interval.ms", intervalMs);
        properties.put("session.timeout.ms", timeoutMs);
        properties.put("auto.offset.reset", autoOffset);
        properties.put("key.deserializer", keyDeserializer);
        properties.put("value.deserializer", valueDeserializer);
        properties.put("enable.auto.commit", autoCommit);
        properties.put("max.poll.records", pollRecords);
        return properties;
    }

    @Bean(initMethod = "init", destroyMethod = "destroy",name = "w_user_brand_provider")
    public HkUserBinlogConsumer kafkaGoodsMessageReceiver(@Qualifier(value = "kafkaUserInfoConsumerProperties") Properties kafkaUserInfoConsumerProperties) {
        if (!switchFlag) {
            return null;
        }
        HkUserBinlogConsumer binlogConsumer = new HkUserBinlogConsumer();
        binlogConsumer.setProperties(kafkaUserInfoConsumerProperties);
        binlogConsumer.setTopics(userBrandProviderTopic);
        return binlogConsumer;
    }
}
